package com.dongdongshop.scoreMq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.service.OrderService;
import com.dongdongshop.service.UserScoreAccountService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "update-user-score",consumerGroup = "my-group-score")
public class scoreConsum implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener{

    @Autowired
    private UserScoreAccountService userScoreAccountService;

    @Autowired
    private RedisTemplate redisTemplate;

    @DubboReference
    private OrderService orderService;

    @Override
    public void onMessage(String s) {
        //玩的,不适合生产开发
    }
    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        //手动接管ACK机制
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取我们的信息
                MessageExt message = list.get(0);
                //获取消息的Id
                String msgId = message.getMsgId();
                //获取消息的实体,并转换成String类型
                String msg = new String(message.getBody());
                JSONObject jsonObject = JSONUtil.parseObj(msg);  //将我们的类型还原成我们传过来之前的样子
                    try {
                        Boolean userScore = redisTemplate.boundHashOps("userScore").hasKey(msgId);
                        if (!userScore){
                            System.out.println("开始消费我们的消息");
                            String orderNum = (String) jsonObject.get("out_trade_no");
                            TbOrder orders =orderService.selectOrderByOrederNum(orderNum);
                            Double points = Double.parseDouble(jsonObject.get("points").toString());
                            Long score = points.longValue();
                            TbUserScoreAccount scoreAccount = new TbUserScoreAccount();
                            scoreAccount.setUserId(orders.getUserId());
                            scoreAccount.setTotalScore(score);
                            userScoreAccountService.updateScoreBy(scoreAccount);

                            redisTemplate.boundHashOps("userScore").put(msgId,"1");
                            redisTemplate.expire("userScore",30, TimeUnit.MINUTES);//设置30分钟的过期时间
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }else {
                            System.out.println("该消息已经消费过了");
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                        int reconsumeTimes = message.getReconsumeTimes();
                        if (reconsumeTimes >3){
                            System.out.println("重试次数过多,手动结束,将数据存入死信队列");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//手动结束,别再试了
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
